package com.gtp.mq.demo.exchange_type.direct;

import java.io.IOException;

import org.apache.commons.lang.SerializationUtils;
import org.junit.Test;

import com.gtp.mq.demo.QueueBindFactory;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class DirectConsumer extends QueueBindFactory{

	@Test
	public void test1() throws Exception {

		// QOS:quality-of-service
		/*
		 * int prefetchSize, ?[!!!rabbitmq还没有实现]
		 * int prefetchCount, 告诉RabbitMQ不要同时给一个消费者推送多于N个消息(QPS)
		 * boolean global      整个渠道还是单个消费者[!!!rabbitmq还没有实现]
		 * 
		 * #官方:
		 * https://www.rabbitmq.com/consumer-prefetch.html
		 */
		channel.basicQos(0, 1, false);

		/***************************** 订阅消费 push模式 **********************/
		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag,
					Envelope envelope, 
					BasicProperties properties, byte[] body)
					throws IOException {

				try {

					Object msg = SerializationUtils.deserialize(body);
					System.out.println("=>msg:" + msg);

					// 非自动ack设置(应答)
					// String routingKey = envelope.getRoutingKey();
					// String contentType = properties.getContentType();
					long deliveryTag = envelope.getDeliveryTag();
					channel.basicAck(deliveryTag, false);
					/*
					 * Q1: basicAck 方法的第二个参数 multiple 取值为 false 时 表示通知 RabbitMQ
					 * 当前消息被确认；如果为 true 则额外将比第一个参数指定的 delivery tag 小的消息一并确认。
					 * 对同一消息的重复确认，【或者对不存在的消息的确认，会产生 IO 异常，导致信道关闭】(未验证)
					 */

					/*
					 * Q2: 忘记关闭会如何 只要程序还在运行，这3条消息就一直是 Unacked 状态，无法被 RabbitMQ
					 * 重新投递。 更厉害的是，RabbitMQ 消息消费并没有超时机制，也就是说，程序不重启，消息就永远是
					 * Unacked 状态。 处理运维事件时不要忘了这些 Unacked 状态的消息。 当程序关闭时（实际只要
					 * Consumer 关闭就行），这3条消息会恢复为 Ready 状态【异常消息未ACK会导致Unacked一直在，阻塞】
					 */

					/*
					 * Q3:
					 * note: 如果这时autoAck=true，那么当consumer接收到相关消息之后
					 * 还没来得及处理就crash掉了，那么这样也算数据丢失【自动ACK提高吞吐量，但是可靠性打折扣】
					 */

					/*
					 * Q4: ##拒收 场景: 当前消费者不能处理，可能需要重新 安排并让其他消费者接收并处理它拒绝接收,不处理,
					 * 可以设置是否放回到队列中还是丢掉
					 * 
					 * #basicReject 一次拒绝一条 #丢弃
					 * channel.basicReject(deliveryTag,false);
					 * 
					 * #重新尝试(切记死循环) 
					 * channel.basicReject(deliveryTag, true);
					 * 
					 * 第一个参数指定 delivery tag
					 * 第二个参数说明如何处理这个失败消息。 
					 * requeue 值为 true  表示该消息重新放回队列头(消息重试)， 
					 * requeue 值为 false 表示放弃这条消息
					 * 
					 * ！！！
					 * 如果没有其他接收者监控这个queue的话，要注意一直无限循环发送的危险
					 * basicNack：可以一次拒绝N条消息
					 * channel.basicNack(deliveryTag,multiple, requeue);
					 * 
					 * #deliveryTag认作消息的偏移量
					 * #若启用如上场景的处理，则建议在 finally中处理
					 */
					/*
					    fix:这里若是能将这类消息,放到其它队列，然后再尝试推送会好很多,异常消息能阻塞全局,
					                    还需要客户端自己处理,如spring的最大重试次数限制等
					 */
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		};

		// 订阅 手动ack
		channel.basicConsume("mydirect_queue_1", false, consumer);
		/***********************************************************/

		/*
		 * 消费方式2:拉取pull模式 GetResponse basicGet(String queue, boolean autoAck)
		 */

		/*
		 * #参数不一致 错误：inequivalent arg 'durable' for exchange 'csExchange' in
		 * vhost '/': received 使用不同的MQ客户端时，常常会出现以上错误信息。 如使用php作为product,使用java,
		 * rubby, python作为consume. 最常见的原因是: durable,
		 * auto_delete,passive参数不一致，保持参数一致性就ok了
		 */

		/*
		 * 封装建议 //状态提前定义好 enum Action { ACCEPT, // 处理成功 RETRY, // 可以重试的错误
		 * REJECT, // 无需重试的错误 }
		 * 
		 * 代码框架如下： Action action = Action.RETRY; try { //
		 * 如果成功完成则action=Action.ACCEPT // 让子类去处理
		 * 
		 * }catch (Exception e) { // 根据异常种类决定是ACCEPT、RETRY还是 REJECT }finally {
		 * // 通过finally块来保证Ack/Nack会且只会执行一次 if (action == Action.ACCEPT) {
		 * channel.basicAck(tag); } else if (action == Action.RETRY) {
		 * channel.basicNack(tag, false, true); } else { channel.basicNack(tag,
		 * false, false); } }
		 */
		
		System.in.read();
		
		channel.close();
	}
}
